package com.smsc.headend.task.engine.executor.impl.collection;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.PaymentSwithPreToPostOptGroupEnum;
import com.smsc.headend.common.enums.errorcode.TaskExecuteErrorCode;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.asset.dmo.ComConfigDMO;
import com.smsc.headend.module.asset.dmo.MeterConfigDMO;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.common.entity.DeviceOperation;
import com.smsc.headend.module.data.dmo.DeviceSAPandKeys;
import com.smsc.headend.module.data.dto.DataItemDTO;
import com.smsc.headend.module.data.enums.DictionaryType;
import com.smsc.headend.module.data.enums.OperationGroup;
import com.smsc.headend.module.data.enums.OperationName;
import com.smsc.headend.module.meterscheme.entity.LoadProfileGroupEntity;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.report.entity.LogScheduleTaskDevice;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.dto.TaskResult;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskExecutorException;
import com.smsc.headend.task.engine.executor.TaskExecutor;
import com.smsc.headend.task.engine.facade.DataItemFacade;
import com.smsc.headend.task.engine.handler.impl.collection.ReadDcuEventProfileHandler;
import com.smsc.headend.task.engine.handler.impl.collection.ReadEventLogBufferHandler;
import com.smsc.headend.task.engine.handler.impl.collection.ReadLoadProfileBufferHandler;
import com.smsc.headend.module.task.dto.AtomicTaskState;
import com.smsc.headend.task.engine.handler.impl.deviceconfig.ReadLpForSwitchPaymentToPostHandler;
import com.smsc.headend.task.engine.service.CollMeterTaskPlanService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.ConnectorService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.collection.CollectTaskService;
import com.smsc.headend.task.engine.service.collection.RecollectTaskService;
import com.smsc.headend.task.engine.service.task.LogScheduleTaskDeviceService;
import com.smsc.headend.task.engine.service.task.TaskPacketService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.CollectionUtils;

import java.nio.charset.Charset;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Slf4j
public class ReadProfileExecutor implements TaskExecutor {
    @Autowired
    TaskPacketService packetService;
    @Autowired
    ConnectorService connectorService;

    @Autowired
    RedisUtils redisUtils;
    @Autowired
    FeignAssetManagementService assetManagementRemoteService;
    @Autowired
    DataItemFacade dataItemFacade;

    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    AssetService assetService;

    @Autowired
    CollectTaskService collectTaskService;

    @Autowired
    RecollectTaskService recollectTaskService;

    @Autowired
    CollMeterTaskPlanService collMeterTaskPlanService;
    @Autowired
    LogScheduleTaskDeviceService logScheduleTaskDeviceService;

    @Override
    public AtomicTaskState execute(Task task, AtomicTask atomicTask) throws UDISTaskExecutorException {
        readProfilePreCheck(task, atomicTask);
        DeviceOperation dictionary = assetService.getDictionary(atomicTask.getOperationId(), DictionaryType.Operation.getType());
        Meter meter = getMeter(task, assetService);
        MeterConfigDMO meterConfigDMO = task.getMeterId() == null ? null : assetService.getMeterConfigById(task.getMeterId());
        ComConfigDMO comConfigDMO = assetService.getComConfigById(task.getComId());
        List<DataItemDTO> dataItemCosemList = dataItemFacade.getDataItemsByOperation(meter, null, dictionary, getParameterMap(task, atomicTask), comConfigDMO == null ? null : comConfigDMO.getModelId());
        if (CollectionUtils.isEmpty(dataItemCosemList)) {
            throw new UDISTaskExecutorException(TaskExecuteErrorCode.DATAITEM_NOT_EXISTS.toString());
        }
        Map<String, Object> paramMap = getParameterMap(task, atomicTask);
        boolean snapshot = false;
        if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType()) || TaskType.ODR_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            String profileType = MapUtil.getStr(paramMap, "profileType");
            if (!StrUtil.isEmpty(profileType) && LoadProfileGroupEntity.ProfileType.Snapshot.getCode() == Integer.parseInt(profileType)) {
                snapshot = true;
            }
        }
        Long taskId = null;
        AtomicTaskState state = null;
        byte[] packets = null;
        if (snapshot) {
            //快照曲线不需要时间范围
            ComEntity terminal = assetService.getComById(task.getComId());
            DataItemDTO dataItemDto = dataItemCosemList.get(0);
            DeviceSAPandKeys deviceKeys = meter != null ? assetManagementRemoteService.getMeterCommunicationKey(meter.getMeterId(), dataItemDto.getRoleId()) :
                    assetManagementRemoteService.getTerminalCommunicationKey(terminal.getComId(), dataItemDto.getRoleId());
            packets = packetService.buildReadOneAttributePacket(dataItemCosemList.get(0), meterConfigDMO == null ? comConfigDMO : meterConfigDMO, deviceKeys);
            Class handlerClass = getHandlerClass(task, atomicTask);
            if (state == null) {
                state = createAtomicState(handlerClass);
            }
        } else {
            state = taskDAO.hashGetAtomicTaskState(atomicTask.getAtomicTaskNo());
            Map<String, Object> initialedMap = collectTaskService.scheduleTaskTimeInit(task, paramMap, meter, task.getTaskType());
            if (task.getTaskType().equals(TaskType.SwitchMeterPaymentPreToPost.toString())) {
                taskId = MapUtil.getLong(paramMap, "taskId");
                initialedMap.put("taskId", taskId);
            }
            Date[] profileStartEnd = collectTaskService.getProfileStartEndDate(task, atomicTask, initialedMap, state);
            Date startDate = profileStartEnd[0];
            Date endDate = profileStartEnd[1];
            if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
                if (logScheduleTaskDeviceService.isLatest(startDate, endDate, MapUtil.getInt(paramMap, "interval"))) {
                    logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Success.getCode(), LogScheduleTaskDevice.StatusDescEnum.LatestProgress.getCode());
                    throw new UDISTaskExecutorException(TaskExecuteErrorCode.SCHEDULE_TASK_IS_LATEST_PROCESS.toString());
                }
            }
            //定抄任务将按配置拆分时间段执行， 未到达最终时间时， handler返回0，不删除atomicTask， 继续传输下一个时间段
            packets = packetService.buildReadProfilePacket(dataItemCosemList, meter, startDate, endDate, meterConfigDMO == null ? comConfigDMO : meterConfigDMO);
            Class handlerClass = getHandlerClass(task, atomicTask);
            if (state == null) {
                state = createAtomicState(handlerClass);
            }
            state.setCollectedEndTv(startDate.getTime() / 1000);
            state.setCollectingEndTv(endDate.getTime() / 1000);
            taskDAO.hashSetAtomicTaskState(atomicTask.getAtomicTaskNo(), task, state.getRetryTimes(), state);
            state.setStored(true);
        }
        log.debug(ByteUtils.byteToHexString(packets));
        if (StrUtil.equalsAny(task.getTaskType(), TaskType.Interface_ODR_Profile.toString(), TaskType.Interface_ODR_LogBook.toString())) {
            // 该task缓存报文
            connectorService.sendPacket(task, atomicTask, dataItemCosemList.get(0).getRoleId(), true, packets);
        } else {
            connectorService.sendPacket(task, atomicTask, dataItemCosemList.get(0).getRoleId(), task.getIsODR(), packets);
        }
        if (task.getTaskType().equals(TaskType.SwitchMeterPaymentPreToPost.toString())) {
            if (atomicTask.getGroupId() == OperationGroup.OnDemandMonthlyBillingProfile.getId().intValue() && atomicTask.getSeq() == 0) {
                assetManagementRemoteService.updateStepStatusToPost(taskId, task.getTaskNo(), task.getMeterId(), PaymentSwithPreToPostOptGroupEnum.ON_DEMAND_MONTHLY_BILLING_PROFILE.getOperationGroup().longValue(), SubTaskStatus.Processing.getCode(), SystemClock.now() / 1000, null, null, false);
            }
        }
        return state;
//        return createAtomicState(ReadLoadProfileBufferHandler.class);
    }

    private boolean readProfilePreCheck(Task task, AtomicTask atomicTask) throws UDISTaskExecutorException {
        if (TaskType.AutoReCollectProfile.toString().equalsIgnoreCase(task.getTaskType())) {
            recollectTaskService.changeDetailIfRequired(task, atomicTask);
        }
        return true;
    }

    private Class getHandlerClass(Task task, AtomicTask atomicTask) {
        if (task.getMeterId() == null) {
            return ReadDcuEventProfileHandler.class;
        } else {
            if (atomicTask.getGroupId() == OperationGroup.ScheduleReadProfiles.getId().intValue()
                    || atomicTask.getGroupId() == OperationGroup.ODRReadProfiles.getId().intValue()) {
                return ReadLoadProfileBufferHandler.class;
            } else if (atomicTask.getGroupId() == OperationGroup.OnDemandMonthlyBillingProfile.getId().intValue()) {
                return ReadLpForSwitchPaymentToPostHandler.class;
            } else {
                return ReadEventLogBufferHandler.class;
            }
        }
    }

    @Override
    public Boolean handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error {}, ", message);
        if (task.getIsODR()) {
            Map<String, Object> param = getParameterMap(task, atomicTask);
            Long profileDataItemId = Long.valueOf(String.valueOf(param.get("dataItemId")));
            // key 回曲线的dataItemId给web
            taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), Convert.toStr(profileDataItemId), message, 3600l, false);
        }
        if (StrUtil.equalsAny(task.getTaskType(), TaskType.Interface_ODR_Profile.name(), TaskType.Interface_ODR_LogBook.name(), TaskType.UAA_MeterReads.name())) {
            // UAA接口的随抄执行失败也要返回给 WF MDM
            TaskResult taskResult = new TaskResult();
            taskResult.setMessageId(task.getMessageId());
            taskResult.setTaskNo(task.getTaskNo());
            Map<String, Object> dataItemMap = JSONUtil.toBean(atomicTask.getAtomicTaskParam(), Map.class);
            Map<String, Object> taskResultMap = JSONUtil.toBean(task.getParamJson(), Map.class);
            taskResultMap.putAll(dataItemMap);
            Meter meter = assetService.getMeterById(task.getMeterId());
            Long profileDataItemId = Long.valueOf(String.valueOf(dataItemMap.get("dataItemId")));
            taskResultMap.put("meterModelId", meter.getModelId());
            taskResultMap.put("meterCustomSn", meter.getCustomerSn());
            taskResultMap.put("profileDataItemId", profileDataItemId);
            taskResult.setExceptionCause(message);
            taskResult.setParamJson(JSONUtil.toJsonStr(taskResultMap));
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));
        }
        if ((TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType()) && !TaskExecuteErrorCode.SCHEDULE_TASK_IS_LATEST_PROCESS.toString().equals(message))
                || TaskType.EventRead_Profile.toString().equalsIgnoreCase(task.getTaskType())
                || TaskType.ScheduleRead_LogBook.toString().equalsIgnoreCase(task.getTaskType())) {
            if (UDISTaskExecutorException.COM_STATUS_EXCEPTION.equalsIgnoreCase(message)) {
                log.warn("readExecutorPauseAsComOffline");
                return false;
            }
            logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.DecodeDataFailed.getCode());
        }

        if (task.getTaskType().equals(TaskType.SwitchMeterPaymentPreToPost.toString())) {
            Map<String, Object> parameterMap = getParameterMap(task, atomicTask);
            Long taskId = MapUtil.getLong(parameterMap, "taskId");
            if (atomicTask.getGroupId() == OperationGroup.OnDemandMonthlyBillingProfile.getId().intValue()) {
                assetManagementRemoteService.updateStepStatusToPost(taskId, task.getTaskNo(), task.getMeterId(), PaymentSwithPreToPostOptGroupEnum.ON_DEMAND_MONTHLY_BILLING_PROFILE.getOperationGroup().longValue(), SubTaskStatus.Failed.getCode(), SystemClock.now() / 1000, SystemClock.now() / 1000, String.format("handle %s fail, [%s]", OperationName.forId(atomicTask.getOperationId()).getName(), message), false);
            }
        }

        return true;
    }
}
